{
 "cells": [
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "from pyalink.alink import *\n",
    "useLocalEnv(1)\n",
    "\n",
    "from utils import *\n",
    "import os\n",
    "import pandas as pd\n",
    "\n",
    "DATA_DIR = ROOT_DIR + \"dog_cat\" + os.sep\n",
    "\n",
    "IMAGE_DIR = DATA_DIR + \"train\" + os.sep\n",
    "\n",
    "TRAIN_96_FILE = \"train_96.ak\"\n",
    "TEST_96_FILE = \"test_96.ak\"\n",
    "\n",
    "TRAIN_32_FILE = \"train_32.ak\"\n",
    "TEST_32_FILE = \"test_32.ak\"\n",
    "\n",
    "MODEL_CNN_FILE = \"model_cnn.ak\"\n",
    "MODEL_EFNET_FILE = \"model_efnet.ak\"\n",
    "MODEL_EFNET_OFFLINE_FILE = \"model_efnet_offline.ak\"\n",
    "\n",
    "PREDICTION_COL = \"pred\"\n",
    "PREDICTION_DETAIL_COL = \"pred_info\""
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_1\n",
    "\n",
    "df = pd.DataFrame(os.listdir(IMAGE_DIR))\n",
    "\n",
    "BatchOperator.fromDataframe(df, schemaStr='relative_path string')\\\n",
    "    .select(\"relative_path, REGEXP_EXTRACT(relative_path, '(dog|cat)') AS label\")\\\n",
    "    .lazyPrint(10)\\\n",
    "    .link(\\\n",
    "        AkSinkBatchOp()\\\n",
    "            .setFilePath(DATA_DIR + \"list_all.ak\")\\\n",
    "            .setOverwriteSink(True)\\\n",
    "    )\n",
    "BatchOperator.execute()\n",
    "\n",
    "splitTrainTestIfNotExist(\n",
    "    AkSourceBatchOp().setFilePath(DATA_DIR + \"list_all.ak\"), \n",
    "    DATA_DIR + \"list_train.ak\", DATA_DIR + \"list_test.ak\", \n",
    "    0.9\n",
    ")\n",
    "\n",
    "AkSourceStreamOp()\\\n",
    "    .setFilePath(DATA_DIR + \"list_train.ak\")\\\n",
    "    .link(\\\n",
    "        ReadImageToTensorStreamOp()\\\n",
    "            .setRelativeFilePathCol(\"relative_path\")\\\n",
    "            .setRootFilePath(IMAGE_DIR)\\\n",
    "            .setImageWidth(32)\\\n",
    "            .setImageHeight(32)\\\n",
    "            .setOutputCol(\"tensor\")\\\n",
    "    )\\\n",
    "    .link(\\\n",
    "        AkSinkStreamOp()\\\n",
    "            .setFilePath(DATA_DIR + TRAIN_32_FILE)\\\n",
    "            .setOverwriteSink(True)\\\n",
    "    )\n",
    "StreamOperator.execute()\n",
    "\n",
    "AkSourceStreamOp()\\\n",
    "    .setFilePath(DATA_DIR + \"list_train.ak\")\\\n",
    "    .link(\\\n",
    "        ReadImageToTensorStreamOp()\\\n",
    "            .setRelativeFilePathCol(\"relative_path\")\\\n",
    "            .setRootFilePath(IMAGE_DIR)\\\n",
    "            .setImageWidth(96)\n",
    "            .setImageHeight(96)\\\n",
    "            .setOutputCol(\"tensor\")\\\n",
    "    )\\\n",
    "    .link(\\\n",
    "        AkSinkStreamOp()\\\n",
    "            .setFilePath(DATA_DIR + TRAIN_96_FILE)\\\n",
    "            .setOverwriteSink(True)\\\n",
    "    )\n",
    "StreamOperator.execute()\n",
    "\n",
    "AkSourceStreamOp()\\\n",
    "    .setFilePath(DATA_DIR + \"list_test.ak\")\\\n",
    "    .link(\\\n",
    "        ReadImageToTensorStreamOp()\\\n",
    "            .setRelativeFilePathCol(\"relative_path\")\\\n",
    "            .setRootFilePath(IMAGE_DIR)\\\n",
    "            .setImageWidth(32)\\\n",
    "            .setImageHeight(32)\\\n",
    "            .setOutputCol(\"tensor\")\\\n",
    "    )\\\n",
    "    .link(\\\n",
    "        AkSinkStreamOp()\\\n",
    "            .setFilePath(DATA_DIR + TEST_32_FILE)\\\n",
    "            .setOverwriteSink(True)\\\n",
    "    )\n",
    "StreamOperator.execute()\n",
    "\n",
    "AkSourceStreamOp()\\\n",
    "    .setFilePath(DATA_DIR + \"list_test.ak\")\\\n",
    "    .link(\\\n",
    "        ReadImageToTensorStreamOp()\\\n",
    "            .setRelativeFilePathCol(\"relative_path\")\\\n",
    "            .setRootFilePath(IMAGE_DIR)\\\n",
    "            .setImageWidth(96)\\\n",
    "            .setImageHeight(96)\\\n",
    "            .setOutputCol(\"tensor\")\\\n",
    "    )\\\n",
    "    .link(\\\n",
    "        AkSinkStreamOp()\\\n",
    "            .setFilePath(DATA_DIR + TEST_96_FILE)\\\n",
    "            .setOverwriteSink(True)\\\n",
    "    )\n",
    "StreamOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2\n",
    "def lr(train_set, test_set) :\n",
    "    Pipeline()\\\n",
    "        .add(\\\n",
    "            TensorToVector()\\\n",
    "                .setSelectedCol(\"tensor\")\\\n",
    "                .setReservedCols([\"label\"])\\\n",
    "        )\\\n",
    "        .add(\\\n",
    "            LogisticRegression()\\\n",
    "                .setVectorCol(\"tensor\")\\\n",
    "                .setLabelCol(\"label\")\\\n",
    "                .setPredictionCol(PREDICTION_COL)\\\n",
    "                .setPredictionDetailCol(PREDICTION_DETAIL_COL)\\\n",
    "        )\\\n",
    "        .fit(train_set)\\\n",
    "        .transform(test_set)\\\n",
    "        .link(\\\n",
    "            EvalBinaryClassBatchOp()\\\n",
    "                .setLabelCol(\"label\")\\\n",
    "                .setPredictionDetailCol(PREDICTION_DETAIL_COL)\\\n",
    "                .lazyPrintMetrics()\\\n",
    "        )\n",
    "    \n",
    "    BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def cnn(train_set, test_set) :\n",
    "    if not(os.path.exists(DATA_DIR + MODEL_CNN_FILE)):\n",
    "        train_set\\\n",
    "            .link(\n",
    "                KerasSequentialClassifierTrainBatchOp()\\\n",
    "                    .setTensorCol(\"tensor\")\\\n",
    "                    .setLabelCol(\"label\")\\\n",
    "                    .setLayers([\n",
    "                        \"Conv2D(32, kernel_size=(3, 3), activation='relu')\",\n",
    "                        \"MaxPooling2D(pool_size=(2, 2))\",\n",
    "                        \"Conv2D(64, kernel_size=(3, 3), activation='relu')\",\n",
    "                        \"MaxPooling2D(pool_size=(2, 2))\",\n",
    "                        \"Flatten()\",\n",
    "                        \"Dropout(0.5)\"\n",
    "                    ])\\\n",
    "                    .setNumEpochs(50)\\\n",
    "                    .setSaveCheckpointsEpochs(2.0)\\\n",
    "                    .setValidationSplit(0.1)\\\n",
    "                    .setSaveBestOnly(True)\\\n",
    "                    .setBestMetric(\"auc\")\\\n",
    "            )\\\n",
    "            .link(\n",
    "                AkSinkBatchOp()\\\n",
    "                    .setFilePath(DATA_DIR + MODEL_CNN_FILE)\\\n",
    "            )\n",
    "        BatchOperator.execute()\n",
    "\n",
    "    KerasSequentialClassifierPredictBatchOp()\\\n",
    "        .setPredictionCol(PREDICTION_COL)\\\n",
    "        .setPredictionDetailCol(PREDICTION_DETAIL_COL)\\\n",
    "        .setReservedCols([\"relative_path\", \"label\"])\\\n",
    "        .linkFrom(\n",
    "            AkSourceBatchOp().setFilePath(DATA_DIR + MODEL_CNN_FILE),\n",
    "            test_set\n",
    "        )\\\n",
    "        .lazyPrint(10)\\\n",
    "        .lazyPrintStatistics()\\\n",
    "        .link(\n",
    "            EvalBinaryClassBatchOp()\\\n",
    "                .setLabelCol(\"label\")\\\n",
    "                .setPredictionDetailCol(PREDICTION_DETAIL_COL)\\\n",
    "                .lazyPrintMetrics()\n",
    "        )\n",
    "    BatchOperator.execute();"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_2\n",
    "sw = Stopwatch()\n",
    "sw.start()\n",
    "\n",
    "AlinkGlobalConfiguration.setPrintProcessInfo(True)\n",
    "\n",
    "train_set = AkSourceBatchOp().setFilePath(DATA_DIR + TRAIN_32_FILE)\n",
    "test_set = AkSourceBatchOp().setFilePath(DATA_DIR + TEST_32_FILE)\n",
    "\n",
    "# lr(train_set, test_set)\n",
    "\n",
    "cnn(train_set, test_set)\n",
    "\n",
    "sw.stop()\n",
    "print(sw.getElapsedTimeSpan())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def efficientnet(train_set, test_set) :\n",
    "    if not(os.path.exists(DATA_DIR + MODEL_EFNET_FILE)):\n",
    "        train_set\\\n",
    "            .link(\n",
    "                KerasSequentialClassifierTrainBatchOp()\\\n",
    "                    .setTensorCol(\"tensor\")\\\n",
    "                    .setLabelCol(\"label\")\\\n",
    "                    .setLayers([\n",
    "                        \"hub.KerasLayer('https://hub.tensorflow.google.cn/google/efficientnet/b0/classification/1')\",\n",
    "                        \"Flatten()\"\n",
    "                    ])\\\n",
    "                    .setNumEpochs(5)\\\n",
    "                    .setIntraOpParallelism(1)\\\n",
    "                    .setSaveCheckpointsEpochs(0.5)\\\n",
    "                    .setValidationSplit(0.1)\\\n",
    "                    .setSaveBestOnly(True)\n",
    "                    .setBestMetric(\"auc\")\n",
    "            )\\\n",
    "            .link(\n",
    "                AkSinkBatchOp()\\\n",
    "                    .setFilePath(DATA_DIR + MODEL_EFNET_FILE)\n",
    "            )\n",
    "        BatchOperator.execute()\n",
    "\n",
    "    KerasSequentialClassifierPredictBatchOp()\\\n",
    "        .setPredictionCol(PREDICTION_COL)\\\n",
    "        .setPredictionDetailCol(PREDICTION_DETAIL_COL)\\\n",
    "        .setReservedCols([\"relative_path\", \"label\"])\\\n",
    "        .linkFrom(\n",
    "            AkSourceBatchOp().setFilePath(DATA_DIR + MODEL_EFNET_FILE),\n",
    "            test_set\n",
    "        )\\\n",
    "        .lazyPrint(10)\\\n",
    "        .lazyPrintStatistics()\\\n",
    "        .link(\n",
    "            EvalBinaryClassBatchOp()\\\n",
    "                .setLabelCol(\"label\")\\\n",
    "                .setPredictionDetailCol(PREDICTION_DETAIL_COL)\\\n",
    "                .lazyPrintMetrics()\n",
    "        )\n",
    "    BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "def efficientnet_offline(train_set, test_set) :\n",
    "    if not(os.path.exists(DATA_DIR + MODEL_EFNET_OFFLINE_FILE)):\n",
    "        train_set\\\n",
    "            .link(\n",
    "                KerasSequentialClassifierTrainBatchOp()\\\n",
    "                    .setTensorCol(\"tensor\")\\\n",
    "                    .setLabelCol(\"label\")\\\n",
    "                    .setLayers([\n",
    "                        \"hub.KerasLayer('\" + DATA_DIR + \"1')\",\n",
    "                        \"Flatten()\"\n",
    "                    ])\\\n",
    "                    .setNumEpochs(5)\\\n",
    "                    .setIntraOpParallelism(1)\\\n",
    "                    .setSaveCheckpointsEpochs(0.5)\\\n",
    "                    .setValidationSplit(0.1)\\\n",
    "                    .setSaveBestOnly(True)\\\n",
    "                    .setBestMetric(\"auc\")\n",
    "            )\\\n",
    "            .link(\n",
    "                AkSinkBatchOp()\\\n",
    "                    .setFilePath(DATA_DIR + MODEL_EFNET_OFFLINE_FILE)\n",
    "            )\n",
    "        BatchOperator.execute()\n",
    "\n",
    "    KerasSequentialClassifierPredictBatchOp()\\\n",
    "        .setPredictionCol(PREDICTION_COL)\\\n",
    "        .setPredictionDetailCol(PREDICTION_DETAIL_COL)\\\n",
    "        .setReservedCols([\"relative_path\", \"label\"])\\\n",
    "        .linkFrom(\n",
    "            AkSourceBatchOp().setFilePath(DATA_DIR + MODEL_EFNET_OFFLINE_FILE),\n",
    "            test_set\n",
    "        )\\\n",
    "        .lazyPrint(10)\\\n",
    "        .lazyPrintStatistics()\\\n",
    "        .link(\n",
    "            EvalBinaryClassBatchOp()\\\n",
    "                .setLabelCol(\"label\")\\\n",
    "                .setPredictionDetailCol(PREDICTION_DETAIL_COL)\\\n",
    "                .lazyPrintMetrics()\n",
    "        )\n",
    "    BatchOperator.execute()"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": [
    "#c_3\n",
    "sw = Stopwatch()\n",
    "sw.start()\n",
    "\n",
    "AlinkGlobalConfiguration.setPrintProcessInfo(True)\n",
    "\n",
    "train_set = AkSourceBatchOp().setFilePath(DATA_DIR + TRAIN_96_FILE)\n",
    "test_set = AkSourceBatchOp().setFilePath(DATA_DIR + TEST_96_FILE)\n",
    "\n",
    "efficientnet(train_set, test_set)\n",
    "\n",
    "efficientnet_offline(train_set, test_set)\n",
    "\n",
    "sw.stop()\n",
    "print(sw.getElapsedTimeSpan())"
   ]
  },
  {
   "cell_type": "code",
   "execution_count": null,
   "metadata": {},
   "outputs": [],
   "source": []
  }
 ],
 "metadata": {
  "kernelspec": {
   "display_name": "Python 3",
   "language": "python",
   "name": "python3"
  },
  "language_info": {
   "codemirror_mode": {
    "name": "ipython",
    "version": 3
   },
   "file_extension": ".py",
   "mimetype": "text/x-python",
   "name": "python",
   "nbconvert_exporter": "python",
   "pygments_lexer": "ipython3",
   "version": "3.8.8"
  }
 },
 "nbformat": 4,
 "nbformat_minor": 4
}
